package org.apache.flink.runtime.metrics.groups;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

import javax.annotation.Nullable;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.metrics.CharacterFilter;
import org.apache.flink.runtime.cleanup.LocallyCleanableResource;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import org.apache.flink.util.concurrent.FutureUtils;

/**
 * @类名: ResourceConsumerMetricGroup
 * @说明: 资源消费者metrics分组
 *
 * @author   leehom
 * @Date	 2023年11月22日 下午2:05:11
 * 修改记录：
 *
 * @see 	 
 */
/**
 * Special {@link org.apache.flink.metrics.MetricGroup} representing a ConsumerMananger.
 *
 * <p>Contains extra logic for adding consumer with resources.
 */
public class ConsumerManagerMetricGroup extends ComponentMetricGroup<ConsumerManagerMetricGroup>
        implements LocallyCleanableResource {

    /** 消费者 */
    private final Map<ConsumerID, ConsumerManagerConsumerMetricGroup> consumers = new HashMap<>();

    /** The name of the consumer represented by this metrics group. */
    @Nullable protected final String hostName;

    public ConsumerManagerMetricGroup(MetricRegistry registry, 
    				            @Nullable String hostName) {
        super(
                registry,
                registry.getScopeFormats().getConsumerManagerFormat().formatScope(hostName),
                null);
        this.hostName = hostName;
    }

    public static ConsumerManagerMetricGroup createResourceConsumerMetricGroup(
            final MetricRegistry metricRegistry,  final String hostName) {
        return new ConsumerManagerMetricGroup(metricRegistry, hostName);
    }

    @Nullable
    public String hostname() {
        return hostName;
    }

    @Override
    protected QueryScopeInfo.ConsumerManagerQueryScopeInfo createQueryServiceMetricInfo(
            CharacterFilter filter) {
        return new QueryScopeInfo.ConsumerManagerQueryScopeInfo();
    }

    // ------------------------------------------------------------------------
    //  增加消费者
    // ------------------------------------------------------------------------
    public ConsumerManagerConsumerMetricGroup addConsumer(ConsumerID consumerId, String consumerName) {
        // get or create a resource metric group
    	ConsumerManagerConsumerMetricGroup currentConsumerGroup;
        synchronized (this) {
            if (!isClosed()) {
                currentConsumerGroup = consumers.get(consumerId);
                // 新建
                if (currentConsumerGroup == null || currentConsumerGroup.isClosed()) {
                    currentConsumerGroup = new ConsumerManagerConsumerMetricGroup(registry, this, consumerId, consumerName);
                    consumers.put(consumerId, currentConsumerGroup);
                }
                return currentConsumerGroup;
            } else {
                return null;
            }
        }
    }

    @Override
    public CompletableFuture<Void> localCleanupAsync(ConsumerID jobId, Executor ignoredExecutor) {
        if (jobId == null) {
            return FutureUtils.completedVoidFuture();
        }
        // 
        synchronized (this) {
        	ConsumerManagerConsumerMetricGroup containedGroup = consumers.remove(jobId);
            if (containedGroup != null) {
                containedGroup.close();
            }
        }
        return FutureUtils.completedVoidFuture();
    }

    public int numRegisteredJobMetricGroups() {
        return consumers.size();
    }

    // ------------------------------------------------------------------------
    //  Component Metric Group Specifics
    // ------------------------------------------------------------------------
    @Override
    protected void putVariables(Map<String, String> variables) {
        variables.put(ScopeFormat.SCOPE_HOST, hostName);
        
    }

    @Override
    protected Iterable<? extends ComponentMetricGroup> subComponents() {
        return consumers.values();
    }

    @Override
    protected String getGroupName(CharacterFilter filter) {
        return "consumermanager";
    }
}
